package org.groupg.project;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.subjects.PublishSubject;
import io.reactivex.rxjava3.subscribers.ResourceSubscriber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class RxJavaDisruptorETL {

    private static final Logger log = LoggerFactory.getLogger(RxJavaDisruptorETL.class);

    static AtomicInteger runCount = new AtomicInteger(1);

    // 1. 定义ETL事件类
    static class EtlEvent {
        private long id;
        private String rawData;
        private String transformedData;
        private boolean processed;
        private Throwable error;

        public void reset() {
            id = 0;
            rawData = null;
            transformedData = null;
            processed = false;
            error = null;
        }
    }

    // 2. 事件工厂
    static class EtlEventFactory implements EventFactory<EtlEvent> {
        @Override
        public EtlEvent newInstance() {
            return new EtlEvent();
        }
    }

    // 3. 事件处理器 - 数据转换
    static class TransformationHandler implements EventHandler<EtlEvent> {
        @Override
        public void onEvent(EtlEvent event, long sequence, boolean endOfBatch) {
            try {
                // 模拟数据转换逻辑
                event.transformedData = event.rawData.toUpperCase() + "_PROCESSED";
                event.processed = true;

                log.info("事件处理器正在处理 - 数据转化：{}",runCount.getAndIncrement());
                // 模拟处理延迟
                TimeUnit.MILLISECONDS.sleep(1);
            } catch (Exception e) {
                event.error = e;
            }
        }
    }

    // 4. RxJava 背压控制器
    static class BackpressureController implements EventHandler<EtlEvent> {
        private final PublishSubject<EtlEvent> subject;
        private final AtomicLong requested = new AtomicLong(0);

        public BackpressureController(PublishSubject<EtlEvent> subject) {
            this.subject = subject;
        }

        @Override
        public void onEvent(EtlEvent event, long sequence, boolean endOfBatch) {
            log.info("事件处理器正在处理 - 背压处理 - 开始");
            // 检查是否有请求
            while (requested.get() <= 0) {
                try {
                    // 等待请求
                    TimeUnit.MILLISECONDS.sleep(1);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
            log.info("事件处理器正在处理 - 背压处理 - 结束");

            // 发送事件到RxJava流水线
            subject.onNext(event);
            requested.decrementAndGet();
        }

        public void request(long n) {
            requested.addAndGet(n);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 配置参数
        final int BUFFER_SIZE = 1024; // Disruptor缓冲区大小
        final int PRODUCER_COUNT = 4; // 生产者线程数
        final int RX_BUFFER_SIZE = 64; // RxJava缓冲区大小

        // 5. 创建Disruptor
        Disruptor<EtlEvent> disruptor = new Disruptor<>(
                new EtlEventFactory(),
                BUFFER_SIZE,
                DaemonThreadFactory.INSTANCE,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        // 6. 创建RxJava背压控制器
        PublishSubject<EtlEvent> rxSubject = PublishSubject.create();
        BackpressureController backpressureController = new BackpressureController(rxSubject);

        // 7. 设置Disruptor处理器链
        disruptor.handleEventsWith(new TransformationHandler())
                .then(backpressureController);

        // 8. 启动Disruptor
        RingBuffer<EtlEvent> ringBuffer = disruptor.start();

        // 9. 创建RxJava处理流水线
        Flowable<EtlEvent> etlPipeline = rxSubject.toFlowable(BackpressureStrategy.BUFFER)
                .onBackpressureBuffer(RX_BUFFER_SIZE,
                        () -> log.info("Rx Buffer Overflow!"),
                        BackpressureOverflowStrategy.DROP_LATEST)
                .observeOn(Schedulers.io(), false, RX_BUFFER_SIZE)
                .doOnNext(event -> {
                    if (event.error != null) {
                        log.error("处理错误: {}", event.error.getMessage());
                    } else if (event.processed) {
                        // 模拟数据加载到目标系统
                        TimeUnit.MILLISECONDS.sleep(5);
                    }
                });

        // 10. 注册背压请求回调
        etlPipeline.subscribe(new ResourceSubscriber<EtlEvent>() {
            @Override
            protected void onStart() {
                request(1); // 初始请求
            }

            @Override
            public void onNext(EtlEvent event) {
                // 处理成功计数
                log.info("处理完成: {} -> {}", event.id, event.transformedData);

                // 请求下一个事件
                request(1);
            }

            @Override
            public void onError(Throwable t) {
                log.error("流水线错误: {}", t.getMessage());
            }

            @Override
            public void onComplete() {
                log.info("ETL处理完成");
            }
        });

        // 11. 创建生产者线程池
        ExecutorService producerPool = Executors.newFixedThreadPool(PRODUCER_COUNT);
        AtomicLong eventCounter = new AtomicLong(1);

        // 12. 启动生产者
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            producerPool.submit(() -> {
                try {
                    while (!Thread.currentThread().isInterrupted()) {
                        long sequence = ringBuffer.next();
                        EtlEvent event = ringBuffer.get(sequence);

                        // 重置事件对象
                        event.reset();

                        // 设置事件数据
                        event.id = eventCounter.getAndIncrement();
                        event.rawData = "原始数据_" + event.id;

                        // 发布事件
                        ringBuffer.publish(sequence);

                        // 控制生产速率
                        TimeUnit.MILLISECONDS.sleep(10);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 13. 运行监控
        new Thread(() -> {
            while (true) {
                try {
                    log.info("\n===== 系统状态监控 =====");
                    log.info("生产事件总数: {}", eventCounter.get() - 1);
                    log.info("Disruptor剩余容量: {}", ringBuffer.remainingCapacity());
                    log.info("RxJava背压请求数: {}", backpressureController.requested.get());

                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();

        // 14. 运行一段时间后停止
        TimeUnit.SECONDS.sleep(30);

        // 15. 关闭资源
        producerPool.shutdownNow();
        disruptor.shutdown();
        rxSubject.onComplete();

        log.info("ETL系统已关闭");
    }
}
